package com.example.pro3_reactor;

import reactor.core.publisher.Flux;
import reactor.core.scheduler.Schedulers;

/**
 * @program: reactivePro
 * @description:
 * @author: zy
 * @create: 2025-04-26 20:22
 */
public class App2_subscrib {
    public static void main(String[] args) {
        Flux< Integer > ints = Flux.range(1, 3);
// 简单订阅:
        ints.subscribe();
        System.out.println("=====================");
// 打印输出
        ints.subscribe(i -> System.out.println(i));
        System.out.println("=====================");
        ints.subscribe(  System.out::println );

        System.out.println("=====================");

        // 返回对象的dispose方法可以取消执行, 所以这里不会有输出
        Flux.range(1,100)
                .publishOn(Schedulers.parallel())  // 异步执行才能看出效果, 否则会同步执行.
                .subscribe(System.out::println)
               .dispose();   // 返回对象的dispose方法可以取消执行,

        System.out.println("=====================");

// 订阅包含错误处理和完成输出
        Flux< Integer > ints2 = Flux.range(1, 4)
                .map(i -> {
                    if (i <= 3) return i;
                    throw new RuntimeException("Got to 4");

                });
        ints2.subscribe(    i -> System.out.println(i),
                error -> System.err.println("出错: " + error),
                () -> System.out.println("完成")

        );


    }
}
